package com.bigdata.quality.config.connector;

import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.bigdata.quality.config.enums.DatabaseEnums;
import com.bigdata.quality.config.exception.CommonException;

import java.util.List;
import java.util.Map;

public class DefaultConnector implements Connector {
  @Override
  public String sourceConnector(ConnectorParams connectorParams) {
    if (connectorParams == null || connectorParams.getDatabaseEnums() == null) {
      throw new CommonException("connectorParams 不能为空");
    }

    if (connectorParams.getDatabaseEnums().equals(DatabaseEnums.HIVE2)) {
      return getFlinkHiveTable(connectorParams);
    }

    return getFlinkColumnSegment(connectorParams) +
        getFlinkSourceSegment(connectorParams);
  }

  @Override
  public String sinkConnector(ConnectorParams connectorParams) {
    return "";
  }

  private String getFlinkHiveTable(ConnectorParams connectorParams) {
    StringBuilder sqlContent = new StringBuilder();
    String dorisJson = connectorParams.getExpandSettings();
    String hiveConfig = "";
    String hiveTableName = connectorParams.getTableName();
    if (StrUtil.isNotBlank(dorisJson)) {
      try {
        JSONObject json = JSONUtil.parseObj(dorisJson);
        hiveConfig = json.get("hiveConfDir").toString();
      } catch (Exception ignored) {
      }
    }

    sqlContent.append("CREATE CATALOG ").append(connectorParams.getDatabaseName()).append(hiveTableName);
    sqlContent.append(" WITH (\n");
    sqlContent.append("'type'='hive',\n");
    sqlContent.append("'default-database'='").append(connectorParams.getDatabaseName()).append("',\n");
    sqlContent.append("'hive-conf-dir'='").append(hiveConfig).append("'\n");
    sqlContent.append(");\n");
    sqlContent.append("USE CATALOG ").append(connectorParams.getDatabaseName()).append(hiveTableName).append(";\n");
    return sqlContent.toString();
  }

  private String getFlinkColumnSegment(ConnectorParams connectorParams) {
    String tableName = connectorParams.getTableName();
    List<String> columnNameList = connectorParams.getColumnNameList();
    List<String> columnFlinkTypeList = connectorParams.getColumnFlinkTypeList();


    StringBuilder sqlContent = new StringBuilder();
    sqlContent.append("CREATE TABLE IF NOT EXISTS ")
        .append("`").append(tableName).append("`").append(" (\n");

    int size = columnNameList.size();
    for (int i = 0; i < columnNameList.size(); i++) {
      String name = columnNameList.get(i);
      String flinkType = columnFlinkTypeList.get(i);

      sqlContent.append("`").append(name).append("` ").append(flinkType);
      if (i < size - 1) {
        sqlContent.append(",\n");
      }
    }

    sqlContent.append(")");
    return sqlContent.toString();
  }

  private String getFlinkSourceSegment(ConnectorParams connectorParams) {
    String tableName = connectorParams.getTableName();
    String username = connectorParams.getUsername();
    String password = connectorParams.getPassword();
    String tenant = connectorParams.getTenant();
    DatabaseEnums databaseEnums = connectorParams.getDatabaseEnums();
    String driverClassName = databaseEnums.getDriverClassName();

    StringBuilder sqlContent = new StringBuilder();
    sqlContent.append(" WITH (\n");
    switch (databaseEnums) {
      case SQL_SERVER:
        sqlContent.append("'connector'='jdbc',\n");
        sqlContent.append("'url'='").append(this.getDataBaseUrl(connectorParams)).append("',\n");
        sqlContent.append("'driver'='").append(driverClassName).append("',\n");
        sqlContent.append("'table-name'='").append(tableName).append("',\n");
        sqlContent.append("'username'='").append(username).append("',\n");
        sqlContent.append("'password'='").append(password).append("'\n");
        break;
      case HIVE2:
        sqlContent.append("'connector'='jdbc',\n");
        sqlContent.append("'table-name'='").append(tableName).append("',\n");
        sqlContent.append("'url'='").append(this.getDataBaseUrl(connectorParams)).append("',\n");
        sqlContent.append("'driver'='").append(driverClassName).append("',\n");
        sqlContent.append("'username'='").append(username).append("',\n");
        sqlContent.append("'password'='").append(password).append("'\n");
        break;
      case DORIS:
        String url = null;
        String dorisJson = connectorParams.getExpandSettings();
        if (StrUtil.isNotBlank(dorisJson)) {
          try {
            JSONObject json = JSONUtil.parseObj(dorisJson);
            url = json.get("fenodes").toString();
          } catch (Exception ignored) {
          }
        }
        sqlContent.append("'connector'='doris',\n");
        if (StrUtil.isNotBlank(url)) {
          sqlContent.append("'fenodes'='").append(url).append("',\n");
        } else {
          sqlContent.append("'fenodes'='").append(connectorParams.getHost()).append(":").append(connectorParams.getPort()).append("',\n");
        }
        sqlContent.append("'table.identifier'='").append(connectorParams.getDatabaseName()).append(".").append(tableName).append("',\n");
        sqlContent.append("'username'='").append(username).append("',\n");
        sqlContent.append("'password'='").append(password).append("'\n");
        break;
      default:
        sqlContent.append("'connector'='jdbc',\n");
        sqlContent.append("'url'='").append(this.getDataBaseUrl(connectorParams)).append("',\n");
        sqlContent.append("'driver'='").append(databaseEnums.getDriverClassName()).append("',\n");
        sqlContent.append("'table-name'='").append(tableName).append("',\n");
        if ("OB".equalsIgnoreCase(databaseEnums.getCode())) {
          if (StrUtil.isBlank(tenant)) {
            throw new CommonException("OceanBase连接、未指定租户信息");
          }
          username = username + "@" + tenant;
          sqlContent.append("'username'='").append(username).append("',\n");
        } else {
          sqlContent.append("'username'='").append(username).append("',\n");
        }
        sqlContent.append("'password'='").append(password).append("'\n");
    }
    sqlContent.append(")");
    return sqlContent.toString();
  }

  private String getDataBaseUrl(ConnectorParams connectorParams) {
    DatabaseEnums databaseEnums = connectorParams.getDatabaseEnums();
    String host = connectorParams.getHost();
    String port = connectorParams.getPort();
    String databaseName = connectorParams.getDatabaseName();

    String url = "";
    switch (databaseEnums) {
      case MYSQL:
        JSONObject inputParams = new JSONObject();
        inputParams.putIfAbsent("useSSL", "false");
        inputParams.putIfAbsent("autoReconnect", "true");
        inputParams.putIfAbsent("serverTimezone", "Asia/Shanghai");
        inputParams.putIfAbsent("nullCatalogMeansCurrent", "true");
        inputParams.putIfAbsent("characterEncoding", "utf-8");
        StringBuilder sb = new StringBuilder();
        for (Map.Entry<String, Object> entry : inputParams.entrySet()) {
          sb.append(entry.getKey()).append("=").append(entry.getValue()).append("&");
        }
        // 移除最后一个 &
        url = "jdbc:mysql://%s:%s/%s?" + sb.substring(0, sb.length() - 1);
        return String.format(url, host, port, databaseName);
      case HIVE2:
        return String.format("jdbc:hive://%s:%s/%s", host, port, databaseName);
      case OCEAN_BASE:
      case GREAT_DB:
      case DORIS:
        return String.format("jdbc:mysql://%s:%s/%s?useSSL=false&autoReconnect=true&serverTimezone=Asia/Shanghai&nullCatalogMeansCurrent=true&characterEncoding=utf-8", host, port, databaseName);
      case SQL_SERVER:
        return "jdbc:sqlserver://" + host + ":" + port + "%3BdatabaseName=" + databaseName;
    }
    return url;
  }
}
